﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Threading;
using System.Diagnostics.Contracts;
using System.Diagnostics;

namespace AzureToolkit.Engine
{
    internal abstract class MessageHandlerBase<TMessage> : IInitializer where TMessage : IAzureQueueMessage
    {
        protected MessageHandlerSettings Settings { get; private set; }
        protected IAzureQueue<TMessage> Queue { get; private set; }

        public MessageHandlerBase(MessageHandlerSettings settings)
        {
            this.Settings = settings;
            this.Queue = new AzureQueue<TMessage>(settings.Account, settings.QueueName, settings.VisibilityTimeout);
        }

        public void Initialize()
        {
            this.Queue.Initialize();
        }

        public void Run()
        {

            // multi-threading
            Task.Factory.StartNew(
              () =>
              {
                  RunWithThrottling(() =>
                  {
                      return this.Cycle();
                  });
              },
              TaskCreationOptions.LongRunning);

            // single thread way
            //while (true)
            //{
            //    var messageCount = this.Queue.RetreiveApproximateMessageCount();
            //    if (messageCount > 0)
            //    {
            //        Cycle();
            //    }
            //    else
            //    {
            //        Thread.Sleep(TimeSpan.FromSeconds(60));
            //    }
            //}
        }

        protected virtual void RunWithThrottling(Func<bool> action)
        {
            while (true)
            {
                var messageCount = this.Queue.RetreiveApproximateMessageCount();
                if (messageCount > 0)
                {
                    int scaleSize = (messageCount / this.Settings.BatchSize) + 1;
                    if (scaleSize > this.Settings.MaxThreads)
                    {
                        scaleSize = this.Settings.MaxThreads;
                    }
                    Parallel.For(0, scaleSize, (i) =>
                    {
                        bool moreWork;
                        do
                        {
                            moreWork = action();
                        } while (moreWork);
                    });
                }
                else
                {
                    Thread.Sleep(TimeSpan.FromSeconds(60));
                }
            }
        }

        protected virtual void ProcessMessages(IEnumerable<TMessage> messages, Action<TMessage> action)
        {
            Contract.Requires(messages != null);
            Contract.Requires(action != null);

            foreach (var message in messages)
            {
                var success = false;
                try
                {
                    action(message);
                    success = true;
                }
                catch (Exception ex)
                {
                    Trace.TraceError(ex.TraceInformation());
                    success = false;
                    if (message.DequeueCount > this.Settings.MaxRetries)
                    {
                        this.Queue.ErrorOutMessage(message, ex);
                    }
                }
                finally
                {
                    if (success || message.DequeueCount > this.Settings.MaxRetries)
                    {
                        this.Queue.DeleteMessage(message);
                    }
                }
            }
        }

        protected abstract bool Cycle();
    }
}
